package com.sanerzone.smscenter.biz.listener;

import com.alibaba.fastjson.JSON;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.sanerzone.smscenter.biz.cache.AccountCacheHelper;
import com.sanerzone.smscenter.biz.message.SMSURREQMessage;
import com.sanerzone.smscenter.biz.message.SMSURRESMessage;
import com.sanerzone.smscenter.biz.utils.DateHelper;
import com.sanerzone.smscenter.biz.utils.MessageExtUtil;
import com.sanerzone.smscenter.biz.utils.StringHelper;
import com.sanerzone.smscenter.common.tools.HttpRequest;
import com.sanerzone.smscenter.common.tools.MQHelper;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.Map;

@Service
public class SMSURREQListener implements MessageListenerConcurrently {
	private Logger logger = LoggerFactory.getLogger(SMSURREQListener.class);

	@Autowired
	public MQHelper mQUtils;
	//推送给用户发送失败的信息
	@Override
	public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {

		Map<String, List<SMSURREQMessage>> map = Maps.newHashMap();
		try {
			for (MessageExt msg : msgs) {
				if (msg.getReconsumeTimes() > 2) {
					return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
				}
				SMSURREQMessage urReqMessage = MessageExtUtil.convertMessageExt(SMSURREQMessage.class, msg);
				if (null == urReqMessage) {
					continue;
				}

				String userId = urReqMessage.getUserid();// 获取用户ID
				String accid = urReqMessage.getAccid();
				//因为这么多的短信过来不是一个用户的，可能是很多的用户的
				if (map.containsKey(userId + "_" + accid)) {
					map.get(userId + "_" + accid).add(urReqMessage);
				} else {
					map.put(userId + "_" + accid, Lists.newArrayList(urReqMessage));
				}
			}
			String result = null;
			Map<String, String	> reportStatusMap = Maps.newHashMap();
			if (map != null && map.size() > 0) {
				for (String key : map.keySet()) {
					String userId = key.split("_")[0];
					String url = AccountCacheHelper.getStringValue(key, "callbackUrl", "");
					//如果url不是以http开头的，就不跳过
					if (!StringHelper.startsWith(url.toLowerCase(), "http")) {
						continue;
					}
					List<SMSURREQMessage> list = map.get(key);
					String param = null;
					param = jsonResult(list);
					try {
						//推送信息给失败用的url(在req中就失败的)
						result = HttpRequest.sendPostJson(url, param, null, 3000);// 推送
					} catch (Exception e) {
						logger.error("用户推送错误:URL:{},参数:{},响应:{}", url, param, result);
						e.printStackTrace();
						return ConsumeConcurrentlyStatus.RECONSUME_LATER;
					}
					logger.info("用户推送:URL:{},参数:{},响应:{}", url, param, result);
					if (StringHelper.isNotBlank(result) && result.length() > 100) {
						result = result.substring(0, 100);
					}
					//这里的list代表着一个用户的多个消息
					for (SMSURREQMessage message : list) {
						reportStatusMap.put(userId+"_"+message.getPhone(),result);
					}
				}
				for (MessageExt msg : msgs) {
					SMSURREQMessage urReqMessage = MessageExtUtil.convertMessageExt(SMSURREQMessage.class, msg);
					SMSURRESMessage uRRESMessage = new SMSURRESMessage();
					uRRESMessage.setBizid(urReqMessage.getBizid());
					String status =reportStatusMap.remove(urReqMessage.getUserid() + "_" +urReqMessage.getPhone());
					uRRESMessage.setResult(status);

					//更新一个pushflag操作
					mQUtils.sendSMSTRES(uRRESMessage.getBizid(), uRRESMessage);
					if (!StringUtils.equals("0",status)) {
						return ConsumeConcurrentlyStatus.RECONSUME_LATER;
					}
				}
			}
		} catch (Exception e) {
			logger.error("{}", e);
		}
		return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
	}

	private String jsonResult(List<SMSURREQMessage> list) {
		Map<String, List<SMSURREQMessage>> map = Maps.newHashMap();
		for (SMSURREQMessage urReqMessage : list) {
			String taskid = urReqMessage.getTaskid();
			if (map.containsKey(taskid)) {
				map.get(taskid).add(urReqMessage);
			} else {
				map.put(taskid, Lists.newArrayList(urReqMessage));
			}
		}

		List<Map<String, String>> rList = Lists.newArrayList();
		Map<String, String> resultMap = null;

		for (String taskid : map.keySet()) {
			List<SMSURREQMessage> resultList = map.get(taskid);
			for (SMSURREQMessage entity : resultList) {
				resultMap = Maps.newHashMap();
				resultMap.put("taskid", taskid);
				resultMap.put("code", entity.getStat());
				if (entity.getStat().startsWith( "F" )){
					resultMap.put("msg", "失败");
				}else if (entity.getStat().startsWith( "T" )){
					resultMap.put("msg", "成功");
				}
				resultMap.put("mobile", entity.getPhone());
				resultMap.put("messageid", entity.getMessageid());
				resultMap.put( "reason",entity.getReserve() );
				resultMap.put("time", DateHelper.getDate("yyyyMMddHHmmss"));
				rList.add(resultMap);
			}
		}
		return JSON.toJSONString(rList);
	}
}
